# coding=utf-8
#下载分别从掘金量化,挖地兔下载数据,需要在这两个网站注册.多线程下载,且已有数据不会重复下载,以此加快下载速度.
from __future__ import print_function, absolute_import
from gm.api import *
import pandas as pd
import time
import datetime
import tushare as ts
from concurrent.futures import *
import tqdm
import chage_data_type
import multiprocessing as mp
import os

print('chage_data_type函数可能导致追加数据时因数据大小不一样而写入失败')
# pro_bar用法,各种数据需要不同积分
set_token('df082f274b3ee380fcd2a98a2e5a7217dac1268b')
pro = ts.pro_api('762be0e261faf0fc1d7807e160da4506b2268d600ce6e0da581cc1b9')
down_day = 600
i = get_instruments(symbols=None, exchanges=None, sec_types='1', names=None, skip_suspended=False, skip_st=False,
                    fields=None, df=True)
i = i[i['trade_date'] - i['listed_date'] >= datetime.timedelta(down_day)]
i = i[~(i['symbol'].str.startswith('SHSE.900') ^ i['symbol'].str.startswith('SZSE.200') ^
        i['symbol'].str.startswith('SHSE.688') ^ i['symbol'].str.startswith('SZSE.300'))]
code_ts = i['symbol']
i = i.set_index('symbol').sort_values('listed_date', ascending=False)
codes = list(i.index)
d = history_n('SZSE.399300', '1d', down_day, end_time=None, fields=None, skip_suspended=True, fill_missing=None,
              adjust=ADJUST_NONE, adjust_end_time='', df=True)['bob'].dt.strftime('%Y-%m-%d')
d_gd = d.tolist()  # .dt.strftime('%Y-%m-%d')
d_ts = list(d.astype('datetime64[ns]').dt.strftime('%Y%m%d'))
# code_ts = [code.split('.', 1) for code in codes]

'''
def run_time(func):
    def wrapper():
        start = time.time()
        func()  # 函数在这里运行
        end = time.time()
        cost_time = end - start
        print("程序耗时 {}".format(cost_time) + ' 秒!')
    return wrapper
'''

def change_day(code):
    code_split = code.split('.', 1)
    return code_split[0]


def change_turn(code):
    code_split = code.split('.', 1)
    return code_split[1]


#@run_time
def turn_ts():
    try:
        print('开始读取turn_daily数据库...')
        s = pd.read_hdf('D:/data/turn_daily.h5', 'turn_daily')
        date_downed = s.groupby('symbol')['trade_date'].agg('max').reset_index()
        date_downed['symbol'] = date_downed['symbol'].apply(change_turn)
        date_downed['undown_date'] = list(d_gd)[-1]
        date_downed['dif'] = date_downed['undown_date'] != date_downed['trade_date']
        need = date_downed[date_downed['dif'] == True]
        list_data = list(need.groupby('symbol'))
        symbols = need.symbol
        '''
        #date_downed['trade_date'] = date_downed['trade_date'].astype('datetime64[ns]')
        date_undown = code_ts.apply(change_turn)
        date_undown = date_undown.reset_index().rename(columns={0: 'symbol'})
        date_undown['undown_date'] = list(d_gd)[-1]
        #date_undown['trade_date'] = date_undown['trade_date'].astype('datetime64[ns]')
        b = pd.merge(date_undown, date_downed, how='outer', on='symbol')
        b['dif'] = b['undown_date'] != b['trade_date']
        need = b[b['dif'] == True]
        need = need.fillna(list(d_gd)[0])#.astype('datetime64[ns]')
        need['date'] = need['date'].astype('datetime64[ns]')
        '''
        print('一共 ' + str(len(need)) + ' 个turnover数据需要更新!')
    except:
        date_undown = code_ts.apply(change_turn)
        date_undown = date_undown.reset_index().rename(columns={0: 'symbol'})
        date_undown['date'] = list(d_gd)[0]
        need = date_undown
        need = need.drop_duplicates()
        print('一共 '+str(len(need))+'个 turnover数据需要下载!')

    def get_turn(df):
        code = df[0]
        data_split = df[1]
        start = str(data_split.loc[code, 'trade_date'].astype('datetime64[ns]') + datetime.timedelta(days=1))
        end = data_split.loc[code, 'undown_date']
        turn = ts.get_hist_data(code, start=start, end=end)
        turn['trade_date'] = turn.index
        turn['symbol'] = code
        turn['symbol'] = turn['symbol'].map(lambda symbol: 'SZSE.' + symbol if symbol.startswith('3') or symbol
                                            .startswith('0') else 'SHSE.' + symbol)
        return turn
        '''
        undown = str(len(need['symbol']) - list(need['symbol']).index(code) - 1)
        #print('还有 ' + undown + ' 个turnover数据正在下载或更新...')
        time.sleep(3)
        new = need.set_index('symbol')
        start = new.loc[code, 'date']
        #next_day = str(datetime.datetime.strptime(start, '%Y-%m-%d')+ datetime.timedelta(days=1)).split(' ', 1)[0]
        next_day = str(start.date() + datetime.timedelta(days=1))#.split(' ', 1)[0]
        end = list(d_gd)[-1]
        turn = ts.get_hist_data(code, start=next_day, end=end)
        '''

    with tqdm.tqdm(total=len(need)) as tq:
        with ThreadPoolExecutor(4) as etr:
            all_task = [etr.submit(get_turn, df) for df in need]
            for future in as_completed(all_task):
                data = future.result()
                try:
                    data = data[['symbol', 'trade_date', 'turnover']]
                    data.to_hdf('turn_daily.h5', 'turn_daily', append=True, format='t', index=False,
                                data_columns=['symbol', 'trade_date'])
                except KeyError:
                    data['turnover'] = 0.0
                    data = data[['symbol', 'trade_date', 'turnover']]
                    data.to_hdf('D:/data/turn_daily.h5', 'turn_daily', append=True, format='t', index=False,
                                data_columns=['symbol', 'trade_date'])
                tq.update()
    print('turnover数据更新完毕!!!')


# @run_time
def daily_ts():
    def get_daily_ts(date):
        time.sleep(1)
        df = pro.daily(trade_date=date)
        df = df[~(df['ts_code'].str.startswith('900') ^ df['ts_code'].str.startswith('200') ^
                  df['ts_code'].str.startswith('688') ^ df['ts_code'].str.startswith('300'))]
        df = df.rename(columns={'ts_code': 'symbol'})
        df['symbol'] = df['symbol'].apply(change_day)
        df['symbol'] = df['symbol'].map(
            lambda code: 'SZSE.' + code if code.startswith('3') or code.startswith('0') else 'SHSE.' + code)
        df['trade_date'] = df['trade_date'].astype('datetime64[ns]')
        df['vol'] = df['vol'].astype('uint64')
        df['amount'] = df['amount'].astype('uint64')
        df = df[['symbol', 'trade_date', 'open', 'high', 'low', 'close', 'pre_close', 'vol', 'amount']]
        df = df.dropna()
        return df
    num = 450
    try:
        print('开始读取本地daily数据库...')
        downed = pd.read_hdf('D:/data/history_daily.h5', 'history_daily', columns=['trade_date']
                             ).trade_date.dt.strftime('%Y-%m-%d').drop_duplicates().tolist()
        dif = list(set(d_gd).difference(set(downed)))
        dif = pd.Series(dif).astype('datetime64[ns]').dt.strftime('%Y%m%d').tolist()
        if len(dif) >= num:
            print('一共 ' + str(len(dif)) + ' 个daily数据需要更新...')
            dif = [dif[n:n + num] for n in range(0, len(dif), num)]
            print('开始分 ' + str(len(dif)) + ' 批下载...')
            for x in dif:
                print('正在下载第 ' + str(dif.index(x) + 1) + ' 批数据,请稍候...')
                all_task = [ThreadPoolExecutor(4).submit(get_daily_ts, date) for date in x]
                with tqdm.tqdm(total=(len(x))) as tq:
                    for future in as_completed(all_task):
                        data = future.result()
                        data.to_hdf('D:/data/history_daily.h5', 'history_daily', append=True, format='t', index=False,
                                    data_columns=['symbol', 'trade_date'])
                        tq.set_description('history ')
                        tq.update()
                print('准备开始下一阶段的下载任务!')
                if len(dif) - dif.index(x) > 1:
                    time.sleep(60)
            print('daily数据更新完毕!!!')
        elif num > len(dif) > 0:
            dif = dif
            print('一共 ' + str(len(dif)) + ' 个daliy数据需要更新...')
            all_task = [ThreadPoolExecutor(4).submit(get_daily_ts, date) for date in dif]
            with tqdm.tqdm(total=(len(dif))) as tq:
                for future in as_completed(all_task):
                    data = future.result()
                    data.to_hdf('D:/data/history_daily.h5', 'history_daily', append=True, format='t', index=False,
                                data_columns=['symbol', 'trade_date'])
                    tq.set_description('history ')
                    tq.update()
            print('daily数据更新完毕!!!')
        else:
            print('daily没有数据需要更新!!!')
    except FileNotFoundError:
        dif = d_ts
        if len(dif) >= num:
            print('一共 ' + str(len(dif)) + ' 个daily数据需要下载...')
            dif = [dif[n:n + num] for n in range(0, len(dif), num)]
            print('开始分 ' + str(len(dif)) + ' 批下载...')
            for x in dif:
                print('正在下载第 ' + str(dif.index(x) + 1) + ' 批数据,请稍候...')
                all_task = [ThreadPoolExecutor(4).submit(get_daily_ts, date) for date in x]
                with tqdm.tqdm(total=(len(x))) as tq:
                    for future in as_completed(all_task):
                        data = future.result()
                        data.to_hdf('D:/data/history_daily.h5', 'history_daily', append=True, format='t', index=False,
                                    data_columns=['symbol', 'trade_date'])
                        tq.set_description('history ')
                        tq.update()
                print('准备开始下一阶段的下载任务!')
                if len(dif) - dif.index(x) > 1:
                    time.sleep(60)
            print('daily数据下载完毕!!!')
        elif num > len(dif) > 0:
            dif = dif
            print('一共 ' + str(len(dif)) + ' 个daliy数据需要更新...')
            all_task = [ThreadPoolExecutor(4).submit(get_daily_ts, date) for date in dif]
            with tqdm.tqdm(total=(len(dif))) as tq:
                for future in as_completed(all_task):
                    data = future.result()
                    data.to_hdf('D:/data/history_daily.h5', 'history_daily', append=True, format='t', index=False,
                                data_columns=['symbol', 'trade_date'])
                    tq.set_description('history ')
                    tq.update()
            print('daily数据下载完毕!!!')
        '''
        all_task = [ThreadPoolExecutor(4).submit(get_daily_ts, date) for date in dif]
        with tqdm.tqdm(total=(len(dif))) as tq:
            for future in as_completed(all_task):
                data = future.result()
                data.to_hdf('D:/data/history_daily.h5', 'history_daily', append=True, format='t', index=False,
                            data_columns=['symbol', 'trade_date'])
                tq.update()
        '''


# @run_time
def instruments():
    def get_ih(x):
        time.sleep(0.5)
        ih = get_history_instruments(codes, fields=None, start_date=x, end_date=x, df=True)
        ih = ih[ih['is_suspended'] == 0]
        ih = ih[['symbol', 'trade_date', 'upper_limit', 'lower_limit', 'sec_level', 'is_suspended', 'pre_close']]
        ih = ih.dropna()
        return ih

    def tpe(dif):
        with ThreadPoolExecutor(8) as etr:
            all_task = [etr.submit(get_ih, x) for x in dif]
            with tqdm.tqdm(total=(len(dif))) as tq:
                for future in as_completed(all_task):
                    data = future.result()
                    data['trade_date'] = data['trade_date'].dt.date.astype('datetime64[ns]')
                    data.to_hdf('D:/data/history_instruments.h5', 'history_instruments', append=True, format='t',
                                index=False,
                                data_columns=['symbol', 'trade_date'])
                    tq.set_description('instruments ')
                    tq.update()

    try:
        print('开始读取本地instruments数据库...')
        downed = pd.read_hdf('D:/data/history_instruments.h5', 'history_instruments',
                             columns=['trade_date']).trade_date.dt.strftime('%Y-%m-%d').drop_duplicates().tolist()
        dif = list(set(d_gd).difference(set(downed)))
        if not dif:
            print('instruments没有数据需要更新!!')
        else:
            print('一共 ' + str(len(dif)) + ' 个instruments数据需要更新...')
            tpe(dif)
        '''
        for x in dif:
            data = get_ih(x)
            data['trade_date'] = data['trade_date'].dt.date.astype(
                'datetime64[ns]')  # .astype('datetime64[ns]')
            data.to_hdf('D:/data/history_instruments.h5', 'history_instruments', append=True, format='t', index=False,
                        data_columns=['symbol', 'trade_date'])
        '''
    except FileNotFoundError:
        dif = d_gd
        print('一共 ' + str(len(dif)) + ' 个instruments数据需要下载...')
        tpe(dif)
        print('instruments数据下载完毕!')


#@run_time
def lbg_ticks():
    store = pd.HDFStore('D:/data/ticks.h5')
    l = [x.split('/', 1)[1] for x in store.keys()]
    try:
        lbg = pd.read_hdf('D:/data/limit_up.h5', 'limit_up')
        lbg = lbg.dropna()
        lbg = lbg[(lbg['ybs'] > 0) & (lbg['sec_level'] == 1)]  # 此处定义取何种日线下载对应的tick
        lbg['trade_date'] = lbg['trade_date'].dt.strftime('%Y-%m-%d')
        lbg = lbg.set_index('trade_date')
        #lbg = lbg[lbg['trade_date'] > datetime.datetime.strptime('2021-02-24', '%Y-%m-%d')].sort_values('trade_date')
        #lbg_gb = lbg.groupby('trade_date')
        lbg_list = lbg.index.drop_duplicates()
    except:
        lbg_list = []
    if not store.keys():
        a = []
    else:
        a = l
    store.close()
    AllCode = lbg_list
    difference = list(set(AllCode).difference(set(a)))
    difference.sort(reverse=True)
    print('一共 ' + str(len(difference)) + ' 个交易日需要下载')
    for x in difference:
        list_tick = []
        df = lbg.loc[x]
        start_time = x.split(' ', 1)[0] + ' 09:24:00'
        end_time = x.split(' ', 1)[0] + ' 15:00:00'
        print(x.split(' ', 1)[0] + ' 一共 ' + str(len(df['symbol'])) + ' 个数据需要下载')
        start_time = x + ' 09:24:00'
        end_time = x + ' 15:00:00'
        #print(x + ' 一共 ' + str(len(df['symbol'])) + ' 个数据需要下载')

        def get_tick(symbol):
            '''
            df1 = df.set_index('symbol')
            time.sleep(0.5)
            upper_limit = df1.loc[symbol, 'upper_limit']
            ytd_t = df1.loc[symbol, 'y_turn']
            ytd_v = df1.loc[symbol, 'ytd_vol']
            ybs = df1.loc[symbol, 'ybs']
            bs = df1.loc[symbol, 'bs']
            o_pct = df1.loc[symbol, 'o_pct']
            '''
            data = history(symbol, 'tick', start_time, end_time, fields=None, adjust=ADJUST_PREV, df=True)
            if not data.empty:
                del data['quotes'], data['open'], data['high'], data['low'], data['last_volume'], data['last_amount']
                '''
                data['upper_limit'] = upper_limit
                data['ytd_t'] = ytd_t
                data['ytd_v'] = ytd_v
                data['up_limit'] = upper_limit
                data['ybs'] = ybs
                data['bs'] = bs
                data['o_pct'] = o_pct
                '''
                data[['cum_volume']] = (data['cum_volume']/100).astype('uint64')
                data['cum_amount'] = data['cum_amount'].astype('uint64')
                #data['now_turn'] = data['cum_volume'] / data['ytd_v'] * data['ytd_t']
                data['created_at'] = data['created_at'].astype('datetime64[ns]') + datetime.timedelta(hours=8)
                #data['price'] = round(data['price'].astype('float32'), 2)
                return data
        all_task = [ThreadPoolExecutor().submit(get_tick, symbol) for symbol in df['symbol']]
        with tqdm.tqdm(total=len(df['symbol'])) as tq:
            for future in as_completed(all_task):
                data = future.result()
                list_tick.append(data)
                tq.set_description('ticks ')
                tq.update()
            if list_tick:
                try:
                    a = pd.concat(list_tick)
                    try:
                        a.to_hdf('D:/data/ticks.h5', x, mode='a')
                        # a.to_hdf('tick.h5', x.split(' ', 1)[0], mode='a')
                    except:
                        print('NaturalNameWarning or PerformanceWarning')
                    print(x + ' 数据已保存')
                except ValueError:
                    print('网站不提供超出三个月的tick数据!!!')
                    break
    print('数据更新完毕!')


#@run_time
def lbg_minutes():
    store = pd.HDFStore('D:/data/minutes.h5')
    l = [x.split('/', 1)[1] for x in store.keys()]
    try:
        # lbg = pd.read_sql('limit_up', sql)
        lbg = pd.read_sql('D:/data/limit_up.h5', 'limit_up')
        lbg = lbg.dropna()
        lbg = lbg[lbg['touch'] == True]  # 此处定义取何种日线下载对应的tick
        lbg = lbg[lbg['trade_date'] > datetime.datetime.strptime('2020-08-24', '%Y-%m-%d')].sort_values('trade_date')
        lbg['trade_date'] = lbg['trade_date'].dt.strftime('%Y-%m-%d')
        lbg = lbg.set_index('trade_date')
        lbg_list = lbg.index.drop_duplicates()
    except:
        lbg_list = []
    if not store.keys():
        a = []
    else:
        a = l
    store.close()
    AllCode = lbg_list
    difference = list(set(AllCode).difference(set(a)))
    difference.sort(reverse=True)
    print('一共 ' + str(len(difference)) + ' 个交易日需要下载')
    for x in difference:
        list_tick = []
        df = lbg.loc[x]
        start_time = x.split(' ', 1)[0] + ' 09:24:00'
        end_time = x.split(' ', 1)[0] + ' 15:00:00'
        print(x.split(' ', 1)[0] + ' 一共 ' + str(len(df['symbol'])) + ' 个数据需要下载')

        # for symbol in df['symbol']:
        def get_minute(symbol):
            time.sleep(1)
            '''
            upper_limit = df.set_index('symbol').loc[symbol, 'upper_limit']
            ytd_t = df.set_index('symbol').loc[symbol, 'ytd_turnover']
            ytd_v = df.set_index('symbol').loc[symbol, 'ytd_volume']
            '''
            data = history(symbol, '60s', start_time, end_time, fields=None, adjust=ADJUST_PREV, df=True)
            if data.empty != True:
                '''
                data['upper_limit'] = upper_limit
                data['ytd_t'] = ytd_t
                data['ytd_v'] = ytd_v
                # data['now_turn'] = data['cum_volume'] / data['ytd_v'] * data['ytd_t'] / 100
                '''
                list_tick.append(data)
                undown = str(len(df['symbol']) - list(df['symbol']).index(symbol))
                # print('还有 ' + undown + ' 个tick数据需要下载...')

        all_task = [ThreadPoolExecutor().submit(get_minute, symbol) for symbol in df['symbol']]
        with tqdm.tqdm(total=len(df['symbol'])) as tq:
            for future in as_completed(all_task):
                data = future.result()
                list_tick.append(data)
                tq.update()
            if list_tick:
                a = pd.concat(list_tick)
                try:
                    a.to_hdf('D:/data/minutes.h5', x.split(' ', 1)[0], mode='w')
                except:
                    print('NaturalNameWarning or PerformanceWarning')
                print(x.split(' ', 1)[0] + ' 数据已保存')
    print('数据更新完毕!')


#@run_time
def one_minutes(df):
    store = pd.HDFStore('D:/data/one_minutes.h5')
    l = [x.split('/', 1)[1] for x in store.keys()]
    try:
        lbg = pd.read_hdf('D:/data/data.h5', 'limit_up')
        lbg = lbg.dropna()
        lbg = lbg[lbg['touch'] == True]  # 此处定义取何种日线下载对应的tick
        lbg = lbg[lbg['trade_date'] > datetime.datetime.strptime('2020-08-24', '%Y-%m-%d')].sort_values('trade_date')
        lbg['trade_date'] = lbg['trade_date'].dt.strftime('%Y-%m-%d')
        lbg = lbg.set_index('trade_date')
        lbg_list = lbg.index.drop_duplicates()
    except:
        lbg_list = []
    if not store.keys():
        a = []
    else:
        a = l
    store.close()


def turn_gm():
    print('turn数据使用掘金数据源,稳定...')
    def get_turn_gm(code):
        time.sleep(1)
        turn = get_fundamentals(table='trading_derivative_indicator', symbols=code, start_date=date,
                                end_date=end,
                                fields='TURNRATE', filter=None, order_by=None, df=True)
        # turn = get_fundamentals_n(table='trading_derivative_indicator', symbols=codes, end_date=end, count=down_day,
        # fields='TURNRATE', filter=None, order_by=None, df=True)
        turn = turn.dropna()
        return turn

    def tpe(codes):
        with tqdm.tqdm(total=len(codes)) as tq:
            with ThreadPoolExecutor(10) as etr:
                all_task = [etr.submit(get_turn_gm, code) for code in codes]
                for future in as_completed(all_task):
                    code = code
                    data = future.result()
                    try:
                        data['trade_date'] = data['pub_date'].dt.strftime('%Y-%m-%d').astype('datetime64[ns]')
                        data = data[['symbol', 'trade_date', 'TURNRATE']]
                        data.to_hdf('D:/data/turn_daily_gm.h5', 'turn_daily_gm', append=True, format='t', index=False,
                                    data_columns=['symbol', 'trade_date'])
                    except KeyError:
                        print(code+'bug!')
                    tq.set_description('turn_gm ')
                    tq.update()

    try:
        print('开始读取本地turnover数据库...')
        data = pd.read_hdf('D:/data/turn_daily_gm.h5', 'turn_daily_gm').dropna().drop_duplicates()
        ih = pd.read_hdf('D:/data/history_instruments.h5', 'history_instruments')
        undown = pd.merge(ih, data, how='left', on=['symbol', 'trade_date'])
        undown = undown[undown['TURNRATE'].isna()]
        max = undown.groupby('symbol')['trade_date'].agg('max').reset_index().rename(columns={'trade_date':'end'})
        min = undown.groupby('symbol')['trade_date'].agg('min').reset_index().rename(columns={'trade_date':'start'})
        undown = pd.merge(max, min, how='outer')
        undown['start'] = undown['start'].dt.date
        undown['end'] = undown['end'].dt.date
        codes = undown.symbol.tolist()
        def get_turn_undown(code):
            start = undown[undown.symbol==code]['start'].iloc[0]
            end = undown[undown.symbol==code]['end'].iloc[0]
            time.sleep(1)
            turn = get_fundamentals(table='trading_derivative_indicator', symbols=code, start_date=start,
                                    end_date=end,
                                    fields='TURNRATE', filter=None, order_by=None, df=True)
            # turn = get_fundamentals_n(table='trading_derivative_indicator', symbols=codes, end_date=end, count=down_day,
            # fields='TURNRATE', filter=None, order_by=None, df=True)
            turn = turn.dropna()
            return turn

        def down_etr(codes):
            print('下载数据库Nan值模式...')
            with tqdm.tqdm(total=len(codes)) as tq:
                with ThreadPoolExecutor(10) as etr:
                    all_task = [etr.submit(get_turn_undown, code) for code in codes]
                    for future in as_completed(all_task):
                        data = future.result()
                        try:
                            data['trade_date'] = data['pub_date'].dt.strftime('%Y-%m-%d').astype('datetime64[ns]')
                            data = data[['symbol', 'trade_date', 'TURNRATE']]
                            data.to_hdf('D:/data/turn_daily_gm.h5', 'turn_daily_gm', append=True, format='t',
                                        index=False,
                                        data_columns=['symbol', 'trade_date'])
                        except KeyError:
                            print('bug!')
                        tq.set_description('turn_gm ')
                        tq.update()

        down_etr(codes)
    except FileNotFoundError:
        date = d_gd[0]
        end = d_gd[-1]
        print('一共 ' + str(len(codes)) + ' 个turnover数据需要下载...')
        print()
        tpe(codes)
        print('turnover数据下载完毕!!!')


def daily_gm():
    print('daily数据使用掘金数据源,稳定...')

    def get_daily(code):
        time.sleep(1)
        daily = history(code, '1d', start_time, end_time, fields=None, skip_suspended=True, fill_missing=None,
                        adjust=ADJUST_PREV, adjust_end_time='', df=True)
        #daily['pre_close'] = daily['close'].shift(-1)
        daily = daily[['symbol', 'open', 'close', 'high', 'low', 'amount', 'volume', 'bob', 'pre_close']]
        daily = daily.dropna()
        return daily

    def tpe(codes):
        with tqdm.tqdm(total=len(codes)) as tq:
            with ThreadPoolExecutor(2) as etr:
                all_task = [etr.submit(get_daily, code) for code in codes]
                for future in as_completed(all_task):
                    data = future.result()
                    data['trade_date'] = data['bob'].dt.strftime('%Y-%m-%d').astype('datetime64[ns]')
                    data.to_hdf('D:/data/daily_gm.h5', 'daily_gm', append=True, format='t', index=False,
                                data_columns=['symbol', 'trade_date'])
                    tq.set_description('daily_gm ')
                    tq.update()

    try:
        print('开始读取本地daily_gm数据库...')
        data = pd.read_hdf('D:/data/daily_gm.h5', 'daily_gm')
        downed = data.trade_date.dt.strftime('%Y-%m-%d').drop_duplicates().tolist()
        downed_symbol = data.symbol.drop_duplicates().tolist()
        dif = list(set(d_gd).difference(set(downed)))
        undown_symbol = list(set(codes).difference(set(downed_symbol)))
        if dif:
            print('一共 '+str(len(dif))+' 个交易日daily_gm数据需要更新...')
            start_time = dif[-1] - datetime.timedelta(days=1)
            end_time = pd.to_datetime(d_gd[-1])
            tpe(codes)
            print('daily_gm数据更新完毕!!!')
        if undown_symbol:
            print('一共 '+str(len(undown_symbol))+' 个股票daily_gm数据需要更新...')
            start_time = pd.to_datetime(d_gd[0]) - datetime.timedelta(days=1)
            end_time = pd.to_datetime(d_gd[-1])
            tpe(undown_symbol)
            print('daily_gm数据更新完毕!!!')
        elif not dif or not undown_symbol:
            print('daily_gm没有数据需要更新')
    except FileNotFoundError:
        start_time = pd.to_datetime(d_gd[0]) - datetime.timedelta(days=1)
        end_time = pd.to_datetime(d_gd[-1])
        print('一共 ' + str(len(codes)) + ' 个daily_gm数据需要下载...')
        tpe(codes)
        print('daliy_gm数据下载完毕!!!')


def sina_big():
    store = pd.HDFStore('D:/data/sina_big.h5')
    l = [x.split('/', 1)[1] for x in store.keys()]
    try:
        lbg = pd.read_hdf('D:/data/limit_up.h5', 'limit_up')
        lbg = lbg.dropna()
        lbg = lbg[(lbg['ybs'] > 0) & (lbg['sec_level'] == 1)]  # 此处定义取何种日线下载对应的tick
        lbg['trade_date'] = lbg['trade_date'].dt.strftime('%Y-%m-%d')
        lbg = lbg.set_index('trade_date')
        #lbg = lbg[lbg['trade_date'] > datetime.datetime.strptime('2021-02-24', '%Y-%m-%d')].sort_values('trade_date')
        #lbg_gb = lbg.groupby('trade_date')
        lbg_list = lbg.index.drop_duplicates()
    except:
        lbg_list = []
    if not store.keys():
        a = []
    else:
        a = l
    store.close()
    AllCode = lbg_list
    difference = list(set(AllCode).difference(set(a)))
    difference.sort(reverse=True)
    print('一共 ' + str(len(difference)) + ' 个交易日需要下载')
    for x in difference:
        list_tick = []
        df = lbg.loc[x]
        print(x.split(' ', 1)[0] + ' 一共 ' + str(len(df['symbol'])) + ' 个数据需要下载')

        def get_tick(symbol):
            code = symbol.split('.', 1)[1]
            data = ts.get_tick_data(code, date=x, src='tt')
            if data:
                data['symbol'] = symbol
                return data
        all_task = [ThreadPoolExecutor().submit(get_tick, symbol) for symbol in df['symbol']]
        with tqdm.tqdm(total=len(df['symbol'])) as tq:
            for future in as_completed(all_task):
                data = future.result()
                list_tick.append(data)
                tq.set_description('sina_big ')
                tq.update()
            if list_tick:
                try:
                    a = pd.concat(list_tick)
                    try:
                        a.to_hdf('D:/data/sina_big.h5', x, mode='a')
                    except:
                        print('NaturalNameWarning or PerformanceWarning')
                    print(x + ' 数据已保存')
                except ValueError:
                    print('网站不提供超出三个月的tick数据!!!')
                    break
    print('数据更新完毕!')


if __name__ == '__main__':
    if not os.path.exists('D:/data'):
        os.mkdir('D:/data')
        print('已在D盘创建data文件夹,下载数据将保存在D:/data文件夹!')
    else:
        print('D盘已存在data文件夹,下载数据将保存在D:/data文件夹!')
    task_daily = mp.Process(target=daily_ts)
    task_instruments = mp.Process(target=instruments)
    task_instruments.start()
    task_daily.start()
    task_instruments.join()
    task_daily.join()
    turn_gm()
    '''
    instruments()
    turn_gm()
    lbg_minutes()
    lbg_ticks()
    sina_big()
'''
